﻿/*
    Copyright (c) 2012, iD Commerce + Logistics
    All rights reserved.

    Redistribution and use in source and binary forms, with or without modification, are permitted
    provided that the following conditions are met:

    Redistributions of source code must retain the above copyright notice, this list of conditions
    and the following disclaimer. Redistributions in binary form must reproduce the above copyright
    notice, this list of conditions and the following disclaimer in the documentation and/or other
    materials provided with the distribution.
 
    THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" AND ANY EXPRESS OR
    IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND
    FITNESS FOR A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR
    CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR
    CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR
    SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
    THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR
    OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
    POSSIBILITY OF SUCH DAMAGE.
*/

using System;
using System.Diagnostics.Contracts;
using System.Collections.Generic;
using System.Linq;
using System.Reactive.Linq;
using System.Reactive.Subjects;
using System.Text;
using System.Threading;
using System.Threading.Tasks;
using System.Reactive.Threading.Tasks;
using System.Collections.Concurrent;
using System.Reactive;
using System.Diagnostics;

namespace IdComLog.Reactive
{
    /// <summary>
    /// Provides extension methods for IObservable.
    /// </summary>
    public static class ObservableExtensions
    {
        /// <summary>
        /// Continues an observable of one type with an observable of another type.
        /// </summary>
        /// <typeparam name="TCurrent">The type of the source observable.</typeparam>
        /// <typeparam name="TNext">The type of the continuation observable.</typeparam>
        /// <param name="src">The source observable.</param>
        /// <param name="next">The continuation observable.</param>
        /// <returns>An observable which starts after the source has finished.</returns>
        public static IObservable<TNext> ContinueWith<TCurrent, TNext>(this IObservable<TCurrent> src, IObservable<TNext> next)
        {
            return src.LastOrDefaultAsync().SelectMany(x => next);
        }

        /// <summary>
        /// Performs an asynchronous action for each item, before passing them through to the next.
        /// </summary>
        /// <typeparam name="T">The type of item to operate on.</typeparam>
        /// <param name="src">The source observable.</param>
        /// <param name="action">The action to perform.</param>
        /// <returns>An observable which performs an action for each item in its antecedent.</returns>
        public static IObservable<T> Do<T>(this IObservable<T> src, Func<T, Task> action)
        {
            Contract.Requires(src != null);
            Contract.Requires(action != null);
            Contract.Ensures(Contract.Result<IObservable<T>>() != null);

            return src
                .Select(x => Observable.FromAsync(async () => { await action(x).ConfigureAwait(false); return x; }))
                .Concat();
        }

        /// <summary>
        /// Filters elements in an observable based on the result of an async function.
        /// </summary>
        /// <typeparam name="T">The type of item to filter.</typeparam>
        /// <param name="src">The enumerable to filter.</param>
        /// <param name="test">The function to test if items should be kept.</param>
        /// <returns>An observable of filtered objects.</returns>
        public static IObservable<T> Where<T>(this IObservable<T> src, Func<T, Task<bool>> test)
        {
            Contract.Requires(src != null);
            Contract.Requires(test != null);
            Contract.Ensures(Contract.Result<IObservable<T>>() != null);

            return src
                .Select(x => Observable.FromAsync(async () => new
                {
                    Item = x,
                    Keep = await test(x)
                }))
                .Concat()
                .Where(x => x.Keep)
                .Select(x => x.Item);
        }

        /// <summary>
        /// Filters elements in an observable based on the result of an async function.
        /// </summary>
        /// <typeparam name="T">The type of item to filter.</typeparam>
        /// <param name="src">The enumerable to filter.</param>
        /// <param name="test">The function to test if items should be kept.</param>
        /// <returns>An observable of filtered objects.</returns>
        public static IObservable<T> Where<T>(this IObservable<T> src, Func<T, int, Task<bool>> test)
        {
            Contract.Requires(src != null);
            Contract.Requires(test != null);
            Contract.Ensures(Contract.Result<IObservable<T>>() != null);

            return src
                .Select((x, i) => Observable.FromAsync(async () => new
                {
                    Item = x,
                    Keep = await test(x, i)
                }))
                .Concat()
                .Where(x => x.Keep)
                .Select(x => x.Item);
        }

        /// <summary>
        /// Allows antecedent observables to continue processing, buffering up to a certain amount of observations.
        /// </summary>
        /// <typeparam name="T">The type to buffer.</typeparam>
        /// <param name="src">The source observable to buffer.</param>
        /// <param name="bufferSize">The maximum amount of observations to buffer.</param>
        /// <returns>An observable which buffers the source.</returns>
        public static IObservable<T> PreBuffer<T>(this IObservable<T> src, int bufferSize)
        {
            Contract.Requires(src != null);
            Contract.Ensures(Contract.Result<IObservable<T>>() != null);

            return Observable.Create<T>(async (o, ct) =>
            {
                using (BlockingCollection<T> queue = new BlockingCollection<T>(bufferSize))
                using (queue.GetConsumingEnumerable(ct).ToObservable().Subscribe(o))
                {
                    await src.Do(x => queue.Add(x, ct)).Finally(() => queue.CompleteAdding());
                }
            });
        }

        /// <summary>
        /// Retries execution of the source observable until it completes successfully, applying a delay between retries.
        /// </summary>
        /// <typeparam name="T">The type to observe.</typeparam>
        /// <param name="src">The observable to retry.</param>
        /// <param name="delay">The delay to wait before retrying.</param>
        /// <returns>An observable which will retry executing its source until successful completion.</returns>
        public static IObservable<T> DelayedRetry<T>(this IObservable<T> src, TimeSpan delay)
        {
            Contract.Requires(src != null);
            Contract.Ensures(Contract.Result<IObservable<T>>() != null);

            if (delay == TimeSpan.Zero) return src.Retry();
            return src.Catch(Observable.Timer(delay).SelectMany(x => src).Retry());
        }

        /// <summary>
        /// Observes a source while a cancellation token has not been triggered.
        /// </summary>
        /// <typeparam name="T">The type of item observed.</typeparam>
        /// <param name="src">The source observable.</param>
        /// <param name="token">The cancellation token.</param>
        /// <returns>A new observable which completes when the given cancellation token is triggered.</returns>
        public static IObservable<T> WhileNotCancelled<T>(this IObservable<T> src, CancellationToken token)
        {
            Contract.Requires(src != null);
            Contract.Ensures(Contract.Result<IObservable<T>>() != null);

            if (!token.CanBeCanceled)
            {
                return src;
            }

            if (token.IsCancellationRequested)
            {
                return Observable.Empty<T>();
            }

            return Observable.Merge(
                token.ToObservable().ContinueWith(Observable.Return(new
                {
                    Value = default(T),
                    IsCancelled = true
                })),
                src.Select(x => new
                {
                    Value = x,
                    IsCancelled = false
                }))
            .TakeWhile(x => !x.IsCancelled)
            .Select(x => x.Value);
        }

        /// <summary>
        /// Inserts an observable if the source observable doesn't generate a value within a due time.
        /// </summary>
        /// <typeparam name="T">The type to observe.</typeparam>
        /// <param name="src">The source observable to operate on.</param>
        /// <param name="dueTime">The maximum time between items from the source.</param>
        /// <param name="func">A function to generate the observable injected when a timeout occurs.</param>
        /// <returns>An observable combining the source and any generated during timeouts.</returns>
        public static IObservable<T> Timeout<T>(this IObservable<T> src, TimeSpan dueTime, Func<IObservable<T>> func)
        {
            Contract.Requires(src != null);
            Contract.Requires(dueTime > TimeSpan.Zero);
            Contract.Requires(func != null);
            Contract.Ensures(Contract.Result<IObservable<T>>() != null);

            return Observable.Create<T>(o =>
            {
                int dueTimeMs = Convert.ToInt32(dueTime.TotalMilliseconds);
                int prevTicks = Environment.TickCount;

                Subject<TickStamped<T>> timerSubject = new Subject<TickStamped<T>>();
                Timer timer = new Timer(obj => ((Subject<TickStamped<T>>)obj).OnNext(new TickStamped<T>()), timerSubject, dueTimeMs, System.Threading.Timeout.Infinite);

                IObservable<TickStamped<T>> stampedSource = src.Select(x => new TickStamped<T> { Tick = Environment.TickCount, Value = x });

                return Observable.Merge(timerSubject, stampedSource)
                    .SelectMany(x =>
                    {
                        if (x.Tick != null)
                        {
                            prevTicks = x.Tick.Value;
                            timer.Change(dueTimeMs, System.Threading.Timeout.Infinite);
                            return Observable.Return(x.Value);
                        }

                        if ((Environment.TickCount - prevTicks) >= dueTimeMs)
                        {
                            return func().Finally(() =>
                            {
                                prevTicks = Environment.TickCount;
                                timer.Change(dueTimeMs, System.Threading.Timeout.Infinite);
                            });
                        }

                        return Observable.Empty<T>();
                    })
                    .Finally(() => timer.Dispose())
                    .Subscribe(o);
            });
        }

        struct TickStamped<T>
        {
            public int? Tick { get; set; }
            public T Value { get; set; }
        }

        public static IObservable<T> MaterializeAndReplay<T>(IObservable<T> obs)
        {
            ReplaySubject<T> s = new ReplaySubject<T>();
            obs.Subscribe(s);
        }
    }
}
